package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.beans.OrderInfo
import com.atguigu.gmall.constances.CommonConstants
import com.atguigu.util.MykafkaUtil.MyKafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.phoenix.spark._

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

object GMVApp extends BaseApp {
  override var appName: String = "GMVApp"
  override var internal: Int = 10

  def main(args: Array[String]): Unit = {
    sc = new StreamingContext("local[4]", appName, Seconds(internal))


//{"payment_way":"1","delivery_address":"gZJhQJWIBHEdCAWiaBHF","consignee":"KcgxWn","create_time":"2021-05-30 17:38:28","order_comment":"gOAmKNJgDxCwBpWOfjDn","expire_time":"","order_status":"2","out_trade_no":"6551962259","tracking_no":"","total_amount":"127.0","user_id":"1","img_url":"","province_id":"4","consignee_tel":"13468129017","trade_body":"","id":"1","parent_order_id":"","operate_time":"2021-05-30 18:28:41"}
    runApp({
      val ds = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_ORDER_INFO, sc)
      //取出kafka中的json,并转换成样例类
      val ds1 = ds.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
        //手机号脱敏
        orderInfo.consignee_tel = orderInfo.consignee_tel.replaceAll("(\\w{3})\\w*(\\w{4})", "$1****$2")
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formatter1 = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        val time = LocalDateTime.parse(orderInfo.create_time,formatter)
       orderInfo.create_hour = time.getHour.toString
        orderInfo.create_date =  time.format(formatter1)
        orderInfo
      })

      ds1.foreachRDD(rdd=>{
        rdd.saveToPhoenix("gmall2020_order_info".toUpperCase(),
          Seq("ID","PROVINCE_ID", "CONSIGNEE", "ORDER_COMMENT", "CONSIGNEE_TEL", "ORDER_STATUS", "PAYMENT_WAY", "USER_ID","IMG_URL", "TOTAL_AMOUNT", "EXPIRE_TIME", "DELIVERY_ADDRESS", "CREATE_TIME","OPERATE_TIME","TRACKING_NO","PARENT_ORDER_ID","OUT_TRADE_NO", "TRADE_BODY", "CREATE_DATE", "CREATE_HOUR"),
          HBaseConfiguration.create(),
          Some("hadoop102:2181"))
      })
    })
  }
}